package com.headfirst.dmp.report.ProvinceCityRpt

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 统计各省市的数据量分布情况
  *
  * 数据基于刚才经过ETL形成的parquet文件  （tips:  parquet文件存和取都依赖于spark-sql）
  *
  * 需求 1：
  * 将统计出来的结果存储成JSON文件 （本次输入为json文件）
  * 需求 2：
  * 将统计出来的结果存储到Mysql中
  */
object ProvinceCityRPT_JSON {

  def main(args: Array[String]): Unit = {

    //1.判断参数个数
    if (args.length != 2) {
      print(
        """
          |com.hedfirst.dmp.tools.Bzip2ParquetV2
          |参数：
          |    logInputPath
          |    ResultOutputPath
        """.stripMargin)
      sys.exit(-1) //非正常退出
    }


    //2.创建sparksession对象
    val session: SparkSession = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .master("local[*]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //设置序列化方式采用KryoSerializer方式（默认的是java序列化）
      .getOrCreate()


    //3.读取parquet文件
    val frame: DataFrame = session.read.parquet(args(0))

    //4.将数据注册成表
    frame.createTempView("v_log")

    //5.执行sql语句，分组统计频次，且排序
    val sql: String = "select provincename,cityname,count(*) ct from v_log group by provincename,cityname order by provincename,cityname"

    val result: DataFrame = session.sql(sql)



    //6.将数据以json格式输出保存

    //判断输出路径是否存在，如果存在，则删除 (测试时用，正式生产环境很少用)
    //使用hadoop的api
    val configuration: Configuration = session.sparkContext.hadoopConfiguration
    val system: FileSystem = FileSystem.get(configuration) //默认配置，读取本地配置文件
    if (system.exists(new Path(args(1)))) {
      system.delete(new Path(args(1)), true) //递归删除
    }

    // coalesce是指将文件合并到1个分区进行输出 （因为分组排序会经历大量shuffle，数据被切分到多个文件输出）
    // 也可以使用result.repartition(1)来设置分区数量为1
    // coalesce的方法本质是Repartition(numPartitions, shuffle = false, logicalPlan)
    // repartition的方法本质是Repartition(numPartitions, shuffle = true, logicalPlan)
    result.coalesce(1).write.json(args(1))

    //7.关闭流
    session.stop()

  }


}
